
package com.lee.utils; 
 
import com.alibaba.fastjson.JSON; 
import com.alibaba.fastjson.JSONArray; 
import com.google.common.collect.Lists; 
import com.lee.enumeration.FPFlinkQualifierEnum; 
import com.lee.enumeration.FinishQualifierEnum;
import com.lee.logger.Logger;
import com.lee.source.HbaseSource;

import java.io.IOException; 
import java.util.ArrayList; 
import java.util.List; 
import java.util.stream.Collectors; 
 
public class DMapUtils { 
    private static String tableName = "marketing:dmap";
    private static String family = "info";
    private static String qualifier = "info";
    private static List<String> indexKey = Lists.newArrayList("$index$");
 
    public static void initMap() throws IOException {
        HbaseSource.truncate(tableName);
        initIndex();
        initItemIndex();
    }
    public static boolean judgeKeyInMap(List<String> key) throws IOException {
        if(key.size() > 1){
            key.sort(String::compareTo);
        }
        return judgeKeyInMap(JSON.toJSONString(key));
 
    }
    public static boolean judgeKeyInMap(String key) throws IOException {
        String result = HbaseSource.get(key,tableName,family,qualifier);
        if(! "".equals(result)){
            return true;
        }else {
            return false;
        }
    }
 
    public static  void addItemToMapGinix(List<String> key,long item) throws IOException {
        key.sort(String::compareTo);
        if (judgeKeyInMap(key)) {
            List<List<Long>> data = getValueByKeyGinix(key);
            List<List<Long>> newData = ArrayUtils.addItemToListGinix(data, item);
            String dataStr = JSON.toJSONString(newData);
            HbaseSource.put(JSON.toJSONString(key), tableName, family, qualifier, dataStr);
 
        } else {
            List<List> data = new ArrayList<>();
            List<Long> newData = new ArrayList<>();
            newData.add(item);
            newData.add(item);
            data.add(newData);
            String dataStr = JSON.toJSONString(data);
            HbaseSource.put(JSON.toJSONString(key), tableName, family, qualifier, dataStr);
        }
 
    }
 
 
 
    public static void setValueByKeyGinix(List<String> key,List<List> value) throws IOException {
        key.sort(String::compareTo);
        String dataStr = JSON.toJSONString(value);
        HbaseSource.put(JSON.toJSONString(key),tableName,family,qualifier,dataStr);
    }
 
    public static List getValueByKeyGinix(List<String> key) throws IOException {
 
        key.sort(String::compareTo);
        return getValueByKeyGinix(JSON.toJSONString(key));
    }
 
 
    public static List getValueByKeyGinix(String key) throws IOException {
        if(judgeKeyInMap(key)) {
            String data = HbaseSource.get(key, tableName, family, qualifier);
            List<String> dataRaw = JSONArray.parseArray(data).toJavaList(String.class);
            List<List<Long>> dataList = dataRaw.stream().map(ritem -> {
                List<Long> dataItem = JSONArray.parseArray(ritem).toJavaList(Long.class);
                return dataItem;
            }).collect(Collectors.toList());
            return dataList;
        } else {
            List<List<Long>> emptyList = new ArrayList<>();
            return emptyList;
        }
 
    }
 
    public static List<String> getAllKey() throws IOException {
        List<String> allKey = HbaseSource.getAllKey(tableName);
        String indexKeyStr = JSON.toJSONString(indexKey);
        allKey = allKey.stream().filter(item-> !indexKeyStr.equals(item)).collect(Collectors.toList());
        return allKey;
    }
 
    public static long getIndex() throws IOException {
        if(judgeKeyInMap(indexKey)){
            List<List<Long>> dataList = getValueByKeyGinix(indexKey);
            return dataList.get(0).get(0);
        } else {
            return 0;
        }
 
    }
    public static long getItemIndex(String qualifierName) throws IOException {
        if(judgeKeyInMap(indexKey)){
            try {
                String data = HbaseSource.get(JSON.toJSONString(indexKey), tableName, family, qualifierName);
                List<String> dataRaw = JSONArray.parseArray(data).toJavaList(String.class);
                List<List<Long>> dataList = dataRaw.stream().map(ritem -> {
                    List<Long> dataItem = JSONArray.parseArray(ritem).toJavaList(Long.class);
                    return dataItem;
                }).collect(Collectors.toList());
                return dataList.get(0).get(0);
            }catch (Exception e){
                return 0;
            }
        } else {
            return 0;
        }
 
    }
 
    public static void addIndex() throws IOException {
        long index = getIndex();
        index += 1;
        List<Long> value = Lists.newArrayList(index);
        List<List> data = new ArrayList<>();
        data.add(value);
        setValueByKeyGinix(indexKey,data);
    }
 
    public static void initIndex() throws IOException {
        List<Long> value = Lists.newArrayList(0L);
        List<List> data = new ArrayList<>();
        data.add(value);
        setValueByKeyGinix(indexKey,data);
    }
 
    public static void initItemIndex() throws IOException {
        List<Long> value = Lists.newArrayList(0L);
        List<List> data = new ArrayList<>();
        data.add(value);
        String dataStr = JSON.toJSONString(data);
        for(int i = 0; i< FinishQualifierEnum.values().length; i++){
            HbaseSource.put(JSON.toJSONString(indexKey),tableName,family, FinishQualifierEnum.values()[i].getCode(),dataStr);
 
        };
        HbaseSource.put(JSON.toJSONString(indexKey),tableName,family, FPFlinkQualifierEnum.DCOUNT.getCode(),dataStr);
 
    }
    public static void setItemIndex(long add,String qualifierName) throws IOException {
        List<Long> value = Lists.newArrayList(add);
        List<List> data = new ArrayList<>();
        data.add(value);
        String dataStr = JSON.toJSONString(data);
 
        HbaseSource.put(JSON.toJSONString(indexKey),tableName,family,qualifierName,dataStr);
    }
 
    public static long getFinishIndex(String qualifierName) throws IOException {
        return getItemIndex(qualifierName);
    }
 
    public static void addFinishIndex(String qualifierName,long finish) throws IOException {
        long finished = getFinishIndex(qualifierName);
        setItemIndex(finish + finished,qualifierName);
    }
 
 
 
 
} 
